package secsmarts;

import java.io.IOException;

import org.apache.hadoop.conf.Configuration;
import org.apache.hadoop.fs.Path;
import org.apache.hadoop.hbase.HBaseConfiguration;
import org.apache.hadoop.hbase.HColumnDescriptor;
import org.apache.hadoop.hbase.HTableDescriptor;
import org.apache.hadoop.hbase.client.HBaseAdmin;
import org.apache.hadoop.hbase.client.Put;
import org.apache.hadoop.hbase.io.ImmutableBytesWritable;
import org.apache.hadoop.hbase.mapreduce.TableOutputFormat;
import org.apache.hadoop.hbase.protobuf.generated.ClientProtos.MutationProto.ColumnValue;
import org.apache.hadoop.io.LongWritable;
import org.apache.hadoop.io.Text;
import org.apache.hadoop.mapreduce.Job;
import org.apache.hadoop.mapreduce.Mapper;
import org.apache.hadoop.mapreduce.Mapper.Context;
import org.apache.hadoop.mapreduce.lib.input.FileInputFormat;

public class LoadFileToHbase2 {
	public static class MapperClass extends Mapper<LongWritable, Text, ImmutableBytesWritable, Put> {
		// 列名
		public static final String[] COLUMNS = { "time", "url", "ip", "country", "code", "type" };

		public void map(LongWritable key, Text value, Context context) throws IOException, InterruptedException {
			String[] cloumnVals = value.toString().split("\t");
			if (cloumnVals.length > 3) {
				String url = cloumnVals[1], ip = cloumnVals[2], host = gethost(url);
				String[] rowkeys = new String[] { url, ip, host };
				for (String rowkey : rowkeys) {
					if (rowkey != null) {
						rowkey = String.format("%03d", rowkey.hashCode() % 1000) + rowkey;
						Put put = new Put(rowkey.getBytes());
						for (int i = 0; i < cloumnVals.length; i++) {
							put.add("info".getBytes(), COLUMNS[i].getBytes(), cloumnVals[i].getBytes());
						}
						context.write(new ImmutableBytesWritable(rowkey.getBytes()), put);
					}
				}
			}
		}

		private String gethost(String url) {
			// TODO Auto-generated method stub
			if (url != null && url.startsWith("http://")) {
				String tmp = url.substring(7);
				return url.substring(0, 7 + tmp.indexOf("/"));
			}
			return null;
		}
	}

	static String tableName = "bingmurls1";

	public static void main(String[] args) {
		try {

			createTable();
			Configuration conf = HBaseConfiguration.create();
			 conf.set("fs.defaultFS", "hdfs://secsmarts");
			conf.set("mapreduce.framework.name", "yarn");
			// conf.set("mapred.job.tracker", "ubuntu:9001");
			conf.set("hbase.zookeeper.quorum", "slave01.secsmarts.com,master.secsmarts.com,slave02.secsmarts.com");
			conf.set("hbase.zookeeper.property.clientPort", "2181");
			conf.set("zookeeper.znode.parent", "/hbase-unsecure");
			Job job = new Job(conf, "file2hbase");
			job.setJarByClass(LoadFileToHbase2.class);
			job.setMapperClass(MapperClass.class);
			job.setNumReduceTasks(0);
			job.setOutputFormatClass(TableOutputFormat.class);
			job.getConfiguration().set(TableOutputFormat.OUTPUT_TABLE, tableName);
			FileInputFormat.addInputPath(job, new Path(args[0]));
			System.out.println(job.waitForCompletion(true) ? 0 : 1);
		} catch (Exception e) {
			e.printStackTrace();
		}
	}

	private static void createTable() throws IOException {
		// TODO Auto-generated method stub

		Configuration conf = HBaseConfiguration.create();
		conf.set("hbase.zookeeper.quorum", "slave01.secsmarts.com,master.secsmarts.com,slave02.secsmarts.com");
		conf.set("zookeeper.znode.parent", "/hbase-unsecure");
		HBaseAdmin admin = new HBaseAdmin(conf);
		if (admin.tableExists(tableName)) {
			return;
		}
		System.out.println("table not exists!creating.......");
		// admin.disableTable(tableName);
		// admin.deleteTable(tableName);
		HTableDescriptor htd = new HTableDescriptor(tableName);
		HColumnDescriptor tcd = new HColumnDescriptor("info");
		htd.addFamily(tcd);// 创建列族
		byte[][] splits = new byte[1000][3];
		for (int i = 0; i < 1000; i++) {
			splits[i] = String.format("%03d", i).getBytes();
		}
		admin.createTable(htd, splits);// 创建表
	}
}
